package com.earthchen.hbase.coprocessor.observer;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * region的协处理器
 *
 * @author: EarthChen
 * @date: 2018/07/29
 */
public class RegionObserverTest extends BaseRegionObserver {

    private byte[] columnFamily = Bytes.toBytes("cf");

    private byte[] countCol = Bytes.toBytes("countCol");

    private byte[] unDeleteCol = Bytes.toBytes("unDeleteCol");

    private RegionCoprocessorEnvironment environment;


    /**
     * 打开region时前执行
     *
     * @param e
     * @throws IOException
     */
    @Override
    public void start(CoprocessorEnvironment e) throws IOException {
        environment = (RegionCoprocessorEnvironment) e;
    }


    /**
     * 关闭region时会执行
     *
     * @param e
     * @throws IOException
     */
    @Override
    public void stop(CoprocessorEnvironment e) throws IOException {
        super.stop(e);
    }


    /**
     * 1. cf:countCol 进行累加，每次插入的时候都要于之前的值进行相加
     *
     * @param e
     * @param put
     * @param edit
     * @param durability
     * @throws IOException
     */
    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        if (put.has(columnFamily, countCol)) {
            // 获取old countcol的value
            Result result = e.getEnvironment().getRegion().get(new Get(put.getRow()));
            int oldNum = 0;
            for (Cell cell : result.rawCells()) {
                if (CellUtil.matchingColumn(cell, columnFamily, countCol)) {
                    oldNum = Integer.valueOf(Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }

            // 获取new  countcol的value
            List<Cell> cellList = put.get(columnFamily, countCol);
            int newNum = 0;
            for (Cell cell : cellList) {
                if (CellUtil.matchingColumn(cell, columnFamily, countCol)) {
                    newNum = Integer.valueOf(Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }

            // 进行累加
            put.addColumn(columnFamily, countCol, Bytes.toBytes(String.valueOf(oldNum + newNum)));
        }
    }

    /**
     * 不能直接删除unDeleteCol
     * <p>
     * 删除countCol的时候unDeleteCol一同删除
     *
     * @param e
     * @param delete
     * @param edit
     * @param durability
     * @throws IOException
     */
    @Override
    public void preDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        // 判断是否操作cf列族
        List<Cell> cellList = delete.getFamilyCellMap().get(columnFamily);

        if (cellList == null || cellList.size() == 0) {
            return;
        }

        boolean deleteFlag = false;
        for (Cell cell : cellList) {
            byte[] qualifier = CellUtil.cloneQualifier(cell);

            if (Arrays.equals(qualifier, unDeleteCol)) {
                throw new IOException("can not delete undel  column");
            }

            if (Arrays.equals(qualifier, countCol)) {
                deleteFlag = true;
            }
        }

        if (deleteFlag) {
            delete.addColumn(columnFamily, unDeleteCol);
        }
    }

}
